package bigdata.hermesfuxi.eagle.etl.jobs; // 定义包名

import bigdata.hermesfuxi.eagle.etl.bean.DataLogBean; // 导入数据日志Bean类
import bigdata.hermesfuxi.eagle.etl.bean.ItemEventCount; // 导入商品事件计数Bean类
import bigdata.hermesfuxi.eagle.etl.utils.FlinkUtils; // 导入Flink工具类
import com.alibaba.fastjson.JSON; // 导入JSON解析工具
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner; // 导入时间戳分配器接口
import org.apache.flink.api.common.eventtime.WatermarkStrategy; // 导入水位线策略类
import org.apache.flink.api.common.functions.AggregateFunction; // 导入聚合函数接口
import org.apache.flink.api.common.functions.RichAggregateFunction; // 导入富聚合函数接口
import org.apache.flink.api.common.serialization.SimpleStringSchema; // 导入简单字符串序列化模式
import org.apache.flink.api.common.state.ListState; // 导入列表状态接口
import org.apache.flink.api.common.state.ListStateDescriptor; // 导入列表状态描述符
import org.apache.flink.api.common.typeinfo.TypeInformation; // 导入类型信息类
import org.apache.flink.api.java.functions.KeySelector; // 导入键选择器接口
import org.apache.flink.api.java.tuple.Tuple3; // 导入三元组类
import org.apache.flink.api.java.tuple.Tuple4; // 导入四元组类
import org.apache.flink.configuration.Configuration; // 导入配置类
import org.apache.flink.streaming.api.datastream.DataStream; // 导入数据流类
import org.apache.flink.streaming.api.datastream.KeyedStream; // 导入键控流类
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; // 导入单输出流操作符类
import org.apache.flink.streaming.api.datastream.WindowedStream; // 导入窗口流类
import org.apache.flink.streaming.api.functions.KeyedProcessFunction; // 导入键控处理函数类
import org.apache.flink.streaming.api.functions.ProcessFunction; // 导入处理函数类
import org.apache.flink.streaming.api.functions.windowing.WindowFunction; // 导入窗口函数接口
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows; // 导入滑动事件时间窗口分配器
import org.apache.flink.streaming.api.windowing.time.Time; // 导入时间类
import org.apache.flink.streaming.api.windowing.windows.TimeWindow; // 导入时间窗口类
import org.apache.flink.util.CollectionUtil; // 导入集合工具类
import org.apache.flink.util.Collector; // 导入收集器接口

import java.time.Duration; // 导入持续时间类
import java.util.List; // 导入列表接口
import java.util.Map; // 导入映射接口

/**
 * 热门商品窗口TopN统计类
 * 功能：统计滑动窗口内各商品的事件数量，并计算TopN热门商品
 */
public class HotItemsWindowTopN {
    /**
     * 主方法，程序入口
     * @param args 命令行参数
     * @throws Exception 异常
     */
    public static void main(String[] args) throws Exception {
        // 从Kafka获取数据源，使用简单字符串序列化模式
        DataStream<String> kafkaSource = FlinkUtils.getKafkaSource(args, SimpleStringSchema.class);
        
        // 处理Kafka数据，将JSON字符串转换为DataLogBean对象
        SingleOutputStreamOperator<DataLogBean> beanStream = kafkaSource.process(new ProcessFunction<String, DataLogBean>() {
            @Override
            public void processElement(String value, Context ctx, Collector<DataLogBean> out) throws Exception {
                // 将JSON字符串解析为DataLogBean对象
                DataLogBean bean = JSON.parseObject(value, DataLogBean.class);
                // 检查bean不为空且事件ID以"product"开头
                if (bean != null && bean.getEventId().startsWith("product")) {
                    // 获取事件属性映射
                    Map map = bean.getProperties();
                    // 检查属性映射包含category_id和product_id字段
                    if (map != null && map.containsKey("category_id") && map.containsKey("product_id")) {
                        // 输出符合条件的bean对象
                        out.collect(bean);
                    }
                }
            }
        });

        // 按照EventTime划分窗口，添加水位线策略
        SingleOutputStreamOperator<DataLogBean> beanStreamWithWaterMark = beanStream.assignTimestampsAndWatermarks(
                // 设置最大乱序时间为5秒的水位线策略
                WatermarkStrategy.<DataLogBean>forBoundedOutOfOrderness(Duration.ofMillis(5000))
                        .withTimestampAssigner(new SerializableTimestampAssigner<DataLogBean>() {
                            @Override
                            public long extractTimestamp(DataLogBean element, long recordTimestamp) {
                                // 从bean对象中提取时间戳作为事件时间
                                return element.getTimestamp();
                            }
                        }));

        // 使用<eventId、categoryId、productId>三元组作为分组键
        KeyedStream<DataLogBean, Tuple3<String, String, String>> keyedStream = beanStreamWithWaterMark.keyBy(new KeySelector<DataLogBean, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> getKey(DataLogBean bean) throws Exception {
                // 获取事件ID
                String eventId = bean.getEventId();
                // 获取属性映射
                Map map = bean.getProperties();
                // 获取分类ID并转换为字符串
                String categoryId = String.valueOf(map.get("category_id"));
                // 获取商品ID并转换为字符串
                String productId = String.valueOf(map.get("product_id"));
                // 返回三元组作为分组键
                return Tuple3.of(eventId, categoryId, productId);
            }
        });

        // 创建滑动窗口：窗口大小20秒，滑动间隔5秒
        WindowedStream<DataLogBean, Tuple3<String, String, String>, TimeWindow> window = keyedStream.window(SlidingEventTimeWindows.of(Time.seconds(20), Time.seconds(5)));

        // 对窗口数据进行聚合计算，使用增量聚合和窗口函数
        SingleOutputStreamOperator<ItemEventCount> aggregateWindow = window.aggregate(new AggregateWindowFunction(), new DataCollectWindowFunction());

        // 按照<eventId、categoryId、start、end>四元组重新分组
        KeyedStream<ItemEventCount, Tuple4<String, String, Long, Long>> tuple2KeyedStream = aggregateWindow.keyBy(new KeySelector<ItemEventCount, Tuple4<String, String, Long, Long>>() {
            @Override
            public Tuple4<String, String, Long, Long> getKey(ItemEventCount value) throws Exception {
                // 返回四元组作为新的分组键
                return Tuple4.of(value.getEventId(), value.getCategoryId(), value.getStart(), value.getEnd());
            }
        });

        // 将topN的商品测流输出 
        
        SingleOutputStreamOperator<ItemEventCount> result = tuple2KeyedStream.process(new HotItemsWindowsTopNFunction(3));

        // 打印结果
        result.print();

        // 执行Flink作业
        FlinkUtils.env.execute();
    }

    /**
     * 窗口增量聚合计算函数（来一条，处理一条）
     * 实现AggregateFunction接口，用于在窗口内进行增量计数
     */
    static class  AggregateWindowFunction implements AggregateFunction<DataLogBean, Long, Long> {
        @Override
        public Long createAccumulator() {
            // 创建累加器，初始值为0
            return 0L;
        }

        @Override
        public Long add(DataLogBean bean, Long accumulator) {
            // 每来一条数据，累加器加1
            return accumulator + 1;
        }

        @Override
        public Long getResult(Long accumulator) {
            // 返回累加器的最终结果
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            // 合并两个累加器（用于窗口合并）
            return a + b;
        }

    }

    /**
     * 数据收集窗口函数
     * 将聚合结果转换为ItemEventCount对象
     */
    static class  DataCollectWindowFunction implements WindowFunction<Long, ItemEventCount, Tuple3<String, String, String>, TimeWindow> {
        @Override
        public void apply(Tuple3<String, String, String> key, TimeWindow window, Iterable<Long> input, Collector<ItemEventCount> out) throws Exception {
            // 获取聚合结果（计数）
            Long count = input.iterator().next();
            // 获取窗口开始时间
            long start = window.getStart();
            // 获取窗口结束时间
            long end = window.getEnd();
            // 创建ItemEventCount对象并输出
            out.collect(new ItemEventCount(key.f0, key.f1, key.f2, count, start, end));
        }
    }


    /**
     * 窗口分组TopN函数
     * 用于计算每个窗口分组的TopN热门商品
     */
    static class HotItemsWindowsTopNFunction extends KeyedProcessFunction<Tuple4<String, String, Long, Long>, ItemEventCount, ItemEventCount> {
        Integer topN; // TopN参数，表示取前N名
        
        /**
         * 构造函数，设置TopN值
         * @param topN 前N名数量
         */
        public HotItemsWindowsTopNFunction(Integer topN) {
            this.topN = topN;
        }

        /**
         * 默认构造函数
         */
        public HotItemsWindowsTopNFunction() {
        }

        private transient ListState<ItemEventCount> listState; // 列表状态，用于存储商品事件计数

        @Override
        public void open(Configuration parameters) throws Exception {
            // 初始化列表状态
            listState = getRuntimeContext().getListState(new ListStateDescriptor<>("aggregateWindowListState", TypeInformation.of(ItemEventCount.class)));
        }

        @Override
        public void processElement(ItemEventCount value, Context ctx, Collector<ItemEventCount> out) throws Exception {
            // 将当前元素添加到状态中
            listState.add(value);

            // 注册定时器，在窗口结束时间+1毫秒时触发
            ctx.timerService().registerEventTimeTimer(value.getEnd() + 1);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<ItemEventCount> out) throws Exception {
            // 获取状态中的所有商品事件计数
            Iterable<ItemEventCount> itemEventCounts = listState.get();
            // 清空状态
            listState.clear();

            // 将迭代器转换为列表
            List<ItemEventCount> list = CollectionUtil.iterableToList(itemEventCounts);
            // 按计数降序排序
            list.sort((a, b) -> b.getCount().compareTo(a.getCount()));

            // 取前topN个结果
            List<ItemEventCount> resultList = list.subList(0, Math.min(topN, list.size()));
            // 输出结果
            for (ItemEventCount itemEventCount : resultList) {
                out.collect(itemEventCount);
            }
        }
    }
}